package com.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gmall.realtime.common.base.BaseApp;
import com.gmall.realtime.common.bean.UserLoginBean;
import com.gmall.realtime.common.constant.Constant;
import com.gmall.realtime.common.function.DorisMapFunction;
import com.gmall.realtime.common.util.DateFormatUtil;
import com.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class DwsUserUserLoginWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsUserUserLoginWindow().start(
                10024,
                4,
                "dws_user_user_login_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    /**
     * 处理数据流的主方法
     *
     * @param env Flink流执行环境
     * @param dataStreamSource 从Kafka读取的原始数据流
     */
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 1. 数据清洗和过滤：将原始JSON字符串转换为JSONObject，并过滤出登录相关数据
        SingleOutputStreamOperator<JSONObject> etlStream = etl(dataStreamSource);
//        etlStream.print();

        // 2. 按用户ID分组处理：计算独立用户数和回流用户数
        SingleOutputStreamOperator<UserLoginBean> keyByStream = getKeyByStream(etlStream);
//        keyByStream.print();

        // 3. 窗口聚合：基于事件时间的10秒滚动窗口，聚合登录指标
        SingleOutputStreamOperator<UserLoginBean> reduceStream = getReduceStream(keyByStream);
//        reduceStream.print();

        // 4. 数据写入Doris：将聚合结果映射为Doris格式并写入
        reduceStream
                .map(new DorisMapFunction<>())
                .sinkTo(
                        FlinkSinkUtil.getDorisSink(
                                Constant.DORIS_DATABASE + "." + Constant.DWS_USER_USER_LOGIN_WINDOW,
                                Constant.DWS_USER_USER_LOGIN_WINDOW
                        )
                );
    }


    /**
     * 获取聚合后的数据流，进行窗口计算
     * 1. 设置水位线和事件时间提取逻辑
     * 2. 定义10秒的滚动事件时间窗口
     * 3. 在窗口内对用户登录数据进行聚合计算
     * 4. 处理窗口结果，添加窗口开始、结束时间等信息
     *
     * @param keyByStream 按用户ID分组后的数据流
     * @return 聚合后的数据流，包含窗口统计结果
     */
    private SingleOutputStreamOperator<UserLoginBean> getReduceStream(SingleOutputStreamOperator<UserLoginBean> keyByStream) {
        return keyByStream
                // 设置水位线策略，允许3秒的乱序时间
                .assignTimestampsAndWatermarks(WatermarkStrategy.<UserLoginBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserLoginBean>() {
                            @Override
                            public long extractTimestamp(UserLoginBean userLoginBean, long l) {
                                // 从UserLoginBean中提取时间戳作为事件时间
                                return userLoginBean.getTs();
                            }
                        }))
                // 定义10秒的滚动事件时间窗口
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 窗口聚合操作
                .reduce(
                        // Reduce函数：对窗口内的数据进行聚合计算
                        new ReduceFunction<UserLoginBean>() {
                            @Override
                            public UserLoginBean reduce(UserLoginBean value1, UserLoginBean value2) throws Exception {
                                // 累加回流用户数
                                value1.setBackCt(value1.getBackCt() + value2.getBackCt());
                                // 累加独立用户数
                                value1.setUuCt(value1.getUuCt() + value2.getUuCt());
                                return value1;
                            }
                        },
                        // 窗口处理函数：添加窗口元数据信息
                        new ProcessAllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {
                            @Override
                            public void process(ProcessAllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>.Context context,
                                                Iterable<UserLoginBean> iterable, Collector<UserLoginBean> collector) throws Exception {
                                // 获取窗口信息
                                TimeWindow window = context.window();
                                long start = window.getStart();
                                long end = window.getEnd();

                                // 获取聚合结果
                                UserLoginBean userLoginBean = iterable.iterator().next();

                                // 设置窗口开始时间、结束时间和当前日期
                                userLoginBean.setStt(DateFormatUtil.tsToDateTime(start));
                                userLoginBean.setEdt(DateFormatUtil.tsToDateTime(end));
                                userLoginBean.setCurDate(DateFormatUtil.tsToDate(System.currentTimeMillis()));

                                // 输出结果
                                collector.collect(userLoginBean);
                            }
                        });
    }


    /**
     * 获取按用户ID分组后的数据流，并计算回流用户数和独立用户数
     * 1. 按用户ID进行分组
     * 2. 使用KeyedProcessFunction处理每个用户的事件
     * 3. 通过状态管理记录用户上次登录日期
     * 4. 计算回流用户(7天未登录)和独立用户(当天首次登录)
     *
     * @param etlStream 经过ETL处理后的JSONObject数据流
     * @return 包含用户登录统计指标(UserLoginBean)的数据流
     */
    private SingleOutputStreamOperator<UserLoginBean> getKeyByStream(SingleOutputStreamOperator<JSONObject> etlStream) {
        // 按用户ID分组处理
        SingleOutputStreamOperator<UserLoginBean> keyByStream = etlStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                // 从common对象中获取用户ID作为分组key
                return jsonObject.getJSONObject("common").getString("uid");
            }
        }).process(new KeyedProcessFunction<String, JSONObject, UserLoginBean>() {
            // 状态变量，用于存储用户上次登录日期
            private ValueState<String> loginDtState;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 初始化状态描述符和状态变量
                loginDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("login_dt", String.class));
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, UserLoginBean>.Context context, Collector<UserLoginBean> collector) throws Exception {
                // 获取事件时间戳
                Long ts = jsonObject.getLong("ts");
                // 将时间戳转换为日期格式
                String curDt = DateFormatUtil.tsToDate(ts);
                // 获取用户上次登录日期
                String value = loginDtState.value();

                // 初始化统计指标
                long backCt = 0L;  // 回流用户数
                long uuCt = 0L;    // 独立用户数

                // 判断是否为当天首次登录
                if (value == null || !value.equals(curDt)) {
                    uuCt = 1L;
                }

                // 判断是否为回流用户(超过7天未登录)
                if (value != null) {
                    if (ts - DateFormatUtil.dateToTs(value) > 7 * 24 * 60 * 60 * 1000) {
                        backCt = 1L;
                    }
                }

                // 更新用户最后登录日期状态
                loginDtState.update(curDt);

                // 如果有统计指标变化，则输出UserLoginBean对象
                if (backCt != 0 || uuCt != 0) {
                    collector.collect(UserLoginBean.builder()
                            .backCt(backCt)
                            .uuCt(uuCt)
                            .ts(ts)
                            .build());
                }
            }
        });
        return keyByStream;
    }


    /**
     * 数据清洗和转换方法
     * 1. 将Kafka原始JSON字符串解析为JSONObject
     * 2. 过滤出用户登录相关数据(包括自动登录和手动登录)
     * 3. 对异常数据进行捕获处理
     *
     * @param kafkaSource 从Kafka读取的原始数据流，每条记录为JSON格式字符串
     * @return 包含过滤后登录数据的JSONObject数据流
     */
    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> kafkaSource) {
        return kafkaSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                try {
                    // 将JSON字符串解析为JSONObject
                    JSONObject jsonObject = JSON.parseObject(s);
                    // 获取common和page节点数据
                    JSONObject common = jsonObject.getJSONObject("common");
                    JSONObject page = jsonObject.getJSONObject("page");

                    // 提取关键字段
                    String uid = common.getString("uid");
                    String ts = jsonObject.getString("ts");
                    String lastPageId = page.getString("last_page_id");

                    // 登录数据过滤逻辑:
                    // 1. uid!=null且last_pageId=null -> 自动登录
                    // 2. uid!=null且last_pageId="login" -> 手动登录
                    if (uid != null && ts != null) {
                        if (lastPageId == null || "login".equals(lastPageId)) {
                            // 符合条件的登录数据加入输出流
                            collector.collect(jsonObject);
                        }
                    }
                } catch (Exception e) {
                    // 异常处理：打印异常堆栈，但不中断流程
                    e.printStackTrace();
                }
            }
        });
    }
}